/*  
  -- The Lime II Project -- 

  A tuplespaces-based middleware for coordinating agents and mobile hosts.
  Copyright (C) 2005.
  Gian Pietro Picco, Amy L. Murphy, Lorenzo Bellini.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
*/

package lime2;
import lime2.ca.ICommunicationAdapter;
import java.util.*;
import java.io.Serializable;

/**
 * Provides (transparent) support for both outgoing and incoming fragmentation.
 *
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini 
 */

public class FragmentationProxy implements ICommunicationAdapter, LimeConstants
{
 // private data members	
 private ICommunicationAdapter dac;	
 private CommunicationMgr cMgr;
 private DefragTable table;
 
 /**
  * Constructor method.
  * 
  * @param dac communication adapter to proxy
  * @param cMgr CommunicationMgr singleton
  */
 FragmentationProxy(ICommunicationAdapter dac, CommunicationMgr cMgr)
 {
  this.dac = dac;
  this.cMgr = cMgr;
  this.table = new DefragTable();
 }
 
 /**
  * This method must be called before the communication adapter is used; it performs all the necessary initializations
  * (i.e. starting thread/s, establishing permanent connections, etc..) 
  *
  * @throws LimeCommunicationException
  */
 public void init() throws LimeCommunicationException
 {
  dac.init();	
  new Thread(new Defragger()).start();
 }
			
 /**
  * Enables this host to join the system. The serializable object may serve to provide startup information for the 
  * joining process (i.e. a host to attach to in a dispatching tree). Please note that communication exceptions are 
  * not thrown; when this method is invoked, it is guaranteed that the LimeServer will join the system as soon as 
  * possible.
  * 
  * @param info startup information for the joining process
  */	
 public void joinSystem(Serializable info) 
 {
  dac.joinSystem(info);	
 }

 /** 
  * Enables this host to leave the system. Please note that communication exceptions are not thrown; when this method
  * is invoked, it is guaranteed that the LimeServer will eventually leave the system. 
  */
 public void leaveSystem() 
 {
  dac.leaveSystem();	
 }

 /**
  * Declares that from now on this node is interested in everything that concerns a specific tuplespace. Please note 
  * that communication exceptions are not thrown; when this method is invoked, it is guaranteed that the LimeServer 
  * will join the group as soon as possible.
  *  
  * @param name name of the tuplespace we are interested in
  */ 
 public void joinGroup(String name) 
 {
  dac.joinGroup(name);	
 }
		 
 /**
  * Declares that from now on this node is uninterested in everything that concerns a specific tuplespace. Please note
  * that communication exceptions are not thrown; when this method is invoked, it is guaranteed that the LimeServer 
  * will eventually leave the group.
  * 
  * @param name name of the tuplespace we are no more interested in
  */ 
 public void leaveGroup(String name)
 {
  dac.leaveGroup(name);	
 }

 /**
  * Sends an operation request to all the group members.
  * 
  * @param name name of the destination tuplespace 
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendOp(String name, RemoteOp d) throws LimeCommunicationException
 {	
  if (FRAGMENT_OUTGOING_ENABLED) 
  {	
   try 
   {	
    FragmentOp fo = new FragmentOp(d);
    if (!fo.tooBig()) dac.sendOp(name, d); 
    else
    {	
     while (fo.hasFragment()) dac.sendOp(name, fo.nextFragment());
     cMgr.getBuffer().extend(DEFRAG_OVERHEAD * fo.fragments);
    }
   }
   catch (FragmentationException fe) { throw new LimeCommunicationException("fragmentation fault", fe, Failures.PARTIAL); }
  }
  else dac.sendOp(name, d);	
 }

 /**
  * Sends an operation request to a specific host.
  * 
  * @param host destination host
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendOp(LimeServerID host, RemoteOp d) throws LimeCommunicationException
 {
  if (FRAGMENT_OUTGOING_ENABLED) 
  {		
   try
   {
    FragmentOp fo = new FragmentOp(d);
    if (!fo.tooBig()) dac.sendOp(host, d); 
    else
    {	
     while (fo.hasFragment()) dac.sendOp(host, fo.nextFragment());
     cMgr.getBuffer().extend(DEFRAG_OVERHEAD * fo.fragments);
    }
   }
   catch (FragmentationException fe) { throw new LimeCommunicationException("fragmentation fault", fe, Failures.PARTIAL); }
  }
  else dac.sendOp(host, d);
 }  	

 /**
  * Sends a reaction registration request to all group members.
  * 
  * @param name name of the destination tuplespace 
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendReaction(String name, RemoteOp d) throws LimeCommunicationException
 {
  if (FRAGMENT_OUTGOING_ENABLED) 
  {		
   try
   {
    FragmentOp fo = new FragmentOp(d);
    if (!fo.tooBig()) dac.sendReaction(name, d); 
    else 
    {
     while (fo.hasFragment()) dac.sendReaction(name, fo.nextFragment());	
     cMgr.getBuffer().extend(DEFRAG_OVERHEAD * fo.fragments);
    }
   }
   catch (FragmentationException fe) { throw new LimeCommunicationException("fragmentation fault", fe, Failures.PARTIAL); }
  }
  else dac.sendReaction(name, d);
 }

 /**
  * Sends a reaction registration request to a specific host.
  * 
  * @param host destination host
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendReaction(LimeServerID host, RemoteOp d) throws LimeCommunicationException
 {
  if (FRAGMENT_OUTGOING_ENABLED) 
  {		
   try
   {
    FragmentOp fo = new FragmentOp(d);
    if (!fo.tooBig()) dac.sendReaction(host, d); 
    else 
    {
     while (fo.hasFragment()) dac.sendReaction(host, fo.nextFragment());	
     cMgr.getBuffer().extend(DEFRAG_OVERHEAD * fo.fragments);
    }
   }
   catch (FragmentationException fe) { throw new LimeCommunicationException("fragmentation fault", fe, Failures.PARTIAL); }
  }
  else dac.sendReaction(host, d);
 } 
	 
 /**
  * Allows to bind a message to a previous operation request (i.e, to gather results).
  * 
  * @param host destination host
  * @param d operation descriptor
  * @throws LimeCommunicationException
  */
 public void sendReply(LimeServerID host, RemoteOp d) throws LimeCommunicationException
 {
  if (FRAGMENT_OUTGOING_ENABLED)
  {	
   try
   {
    FragmentOp fo = new FragmentOp(d);
    if (!fo.tooBig()) dac.sendReply(host, d);  
    else 
    {
     while (fo.hasFragment()) dac.sendReply(host, fo.nextFragment());	
     cMgr.getBuffer().extend(DEFRAG_OVERHEAD * fo.fragments);
    }
   }
   catch (FragmentationException fe) { throw new LimeCommunicationException("fragmentation fault", fe, Failures.PARTIAL); }
  }
  else dac.sendReply(host, d);  
 } 
	
 /**
  * Returns a received RemoteOp (FIFO order, blocking).
  * 
  * @return RemoteOp
  */
 public RemoteOp getNextOp()
 {
  RemoteOp ret = dac.getNextOp();
  if (ret.type == LimeConstants.FRAGMENT) 
  {
   if (DEFRAGMENT_INCOMING_ENABLED) 
   {
	table.collect((FragmentOp) ret);
	// if this fragment is part of a reconciliation process all threads waiting for replies from that tuplespace must
	// have their timeouts renewed; otherwise if this fragment is part of a non-reconciliation operation already in 
	// progress we renew only the timeout of the specific waiting thread  
	if (((FragmentOp) ret).isReconcileOp()) cMgr.renewAll(ret.name);
	else
    if (ret.targetT != -1) cMgr.getBuffer(ret.targetT).renew();	
   }
   return getNextOp();
  }
  else return ret;
 }

 /**
  * Deactivates the communication layer.
  */
 public void shutdown()
 {
  dac.shutdown();
 }
 
 /**
  * Returns whether an host is engaged.
  * 
  * @param host
  * @return <code>true</code> or <code>false</code>
  */
 public boolean queryHost(LimeServerID host)
 {
  return dac.queryHost(host);
 }
 
 /**
  * Adds a new (discovered) host.
  * 
  * @param host
  */
 public void addHost(LimeServerID host)
 {
  dac.addHost(host);
 }
  
 /**
  * Removes a (probable) dead host.
  * 
  * @param host
  */
 public void removeHost(LimeServerID host)
 {
  dac.removeHost(host);
 }
 
 /**
  * Models a defrag entry within the defrag table. A defrag entry identifies a re-assembly process. 
  */
 
 private class DefragEntry implements HashID
 { 
  // private data members
  private long ti;             // operation identifier   
  private long time;           // timestamp (= arrival of last fragment)
  private int seqnum;          // sequence number
  private int fragments;       // expected number of fragments
  private LimeServerID source; // source host

  /**
   * Constructor method.
   * 
   * @param fo FragmentOp to convert into an entry
   */ 
  DefragEntry(FragmentOp fo)
  {
   this.ti = fo.ti;
   this.seqnum = fo.seqnum;
   this.source = fo.source;
   this.fragments = fo.fragments;
   this.time = new Date().getTime() + REASSEMBLY_TIMEOUT;
  }
  
  /**
   * Refreshes the timestamp of an entry. Invoked when a new fragment is collected.
   */
  void refresh()
  {
   time = new Date().getTime() + REASSEMBLY_TIMEOUT;	
  }
  
  /**
   * Tests for equality.
   * 
   * @param o Object
   * @return <code>true</code> if the objects match, <code>false</code> otherwise 
   */ 
  public boolean equals(Object o)
  {
   boolean ret = false;	
   if (o instanceof DefragEntry)
   {
    DefragEntry de = (DefragEntry) o;
    if (ti == de.ti && seqnum == de.seqnum && source.equals(de.source)) ret = true;
   }
   return ret;
  }
  
  /**
   * Returns the hashcode.
   * 
   * @return hashcode 
   */
  public int hashCode()
  {
   return (int) (ti + seqnum + source.hashCode());  	
  }
  
 }
  
 /**
  * Models the defrag table. The defrag table stores for each defrag entry the corresponding defrag buffer. The 
  * defrag buffer collects the fragments of the remote op of a specific operation. As soon as a fragment arrives the 
  * timestamp of the entry is refreshed. Please note that each remote operation can have only one entry within the 
  * table. 
  */
 
 private class DefragTable
 {
  // private data members (please note that java.util.Hashtable is internally synchronized)
  private Hashtable dt = new Hashtable(); // structure: [entry,buffer]
  private Hashtable dk = new Hashtable(); // structure: [entry,entry]
  
  /**
   * Collects a FragmentOp. If there is no matching defrag entry then a new pair (defrag entry - defrag buffer) is 
   * inserted in the table; the given FragmentOp is stored within the corresponding defrag buffer (and the timestamp
   * of the entry is refreshed). Duplicated fragments are discarded. 
   * 
   * @param fo FragmentOp to collect
   */
  void collect(FragmentOp fo)
  {
   DefragEntry key = new DefragEntry(fo);	
   Vector v = (Vector) dt.get(key);	
   if (v == null) { v = new Vector(); synchronized(this) { dt.put(key, v); dk.put(key, key); } } 
   if (!v.contains(fo)) 
   {
   	v.addElement(fo);
    ((DefragEntry) dk.get(key)).refresh();
   } 
  }
    
  /**
   * Returns an enumeration of all the defrag entries for this table.
   * 
   * @return enumeration of the defrag entries
   */
  Enumeration entries()
  {
   return dt.keys();
  }
  
  /**
   * Returns whether all the fragments have arrived.
   *
   * @param de defrag entry
   * @return <code>true</code> or <code>false</code>
   */
  boolean complete(DefragEntry de)
  {
   if (de.fragments == ((Vector) dt.get(de)).size()) return true; else return false;	
  }
  
  /**
   * Returns whether the timestamp for the given entry is already expired or not.
   *
   * @param de defrag entry
   * @return <code>true</code> or <code>false</code>
   */
  boolean expired(DefragEntry de)
  {
   if (new Date().getTime() - de.time > REASSEMBLY_TIMEOUT) return true; else return false;  	
  }
 
  /**
   * Removes an entry from the table.
   * 
   * @param de defrag entry to remove
   */
  void remove(DefragEntry de)
  {
   synchronized(this) { dt.remove(de); dk.remove(de); }
  }
  
  /**
   * Starts the defrag process on the given entry.
   * 
   * @param defrag entry
   * @return re-assembled remote op 
   */ 
  RemoteOp defrag(DefragEntry de) throws FragmentationException
  {
   Vector v = (Vector) dt.get(de);	
   if (v == null) throw new FragmentationException("defragmentation fault");
   FragmentOp ret = null;	
   boolean ok = true;	

   ret = removeFragment(0, v);
   if (ret != null)  
   {	
    for (int i = 1; ok && i < ret.fragments; i++)
    {
   	 FragmentOp next = removeFragment(i, v);
  	 if (next != null) ret.mergeFragment(next); else ok = false; 
    }
   }
   else ok = false;
    
   if (!ok) throw new FragmentationException("defragmentation fault");
   else return ret.reAssemble(); 
  }
  
  // removes the fragment with offset "n" from the given defrag buffer (Vector "v")
  private FragmentOp removeFragment(int n, Vector v)
  {
   int i = 0;
   while (i < v.size()) if (((FragmentOp) v.get(i)).offset == n) return (FragmentOp) v.remove(i); else i++; 
   return null;
  }
  
 }
 
 /**
  * Models the defragger thread. This thread periodically wakes up (see LimeConstants) and inspects the entries of 
  * the defrag table. If an entry is expired it removes it; otherwise, if all the fragments are arrived, it tries to
  * re-assembly the remote op.
  */
 
 private class Defragger implements Runnable
 {
  // private constant	
  private final long Isleep = REASSEMBLY_TIMEOUT / DEFRAG_SCHEDULED;	
 
  /**
   * Run method.
   */	
  public void run()
  {
   // loop forever	
   for (;;)
   {	
   	// sleeps
    try { Thread.sleep(Isleep); } catch (InterruptedException ie) { ie.printStackTrace(); }
    // wakes up
    Enumeration e = table.entries();
    while (e.hasMoreElements())
    {
     DefragEntry de = (DefragEntry) e.nextElement();
     // all the fragments of this op have arrived ?
     if (table.complete(de)) 
     {	
      try 
	  {	
       RemoteOp ret = table.defrag(de); 
       table.remove(de);
       CommunicationMgr.get().dispatch(ret);
      }
      catch (FragmentationException fe) { fe.printStackTrace(); }
	 }
     else if (table.expired(de)) table.remove(de); // if the reassembly timeout is expired, kill the entry
    } 
   } 
   // end loop 
  }
  
 }

@Override
public HashSet<LimeServerID> multicast(LimeServerID[] hosts, RemoteOp d)
		throws LimeCommunicationException {
	return null;
	
}

@Override
public HashSet<LimeServerID> multicast(RemoteOp d) throws LimeCommunicationException {
	return null;
}
  
}